package com.atguigu.chapter11;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink07_SQL_BaseUse_1 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_1", 6000L, 60)
        );

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Table table = tEnv.fromDataStream(stream);


        // 1. 查询未注册的表  表没有名字

        // 缺点是没返回值，可以执行ddl语句也就是没有返回值的sql 增删改语句
//        tEnv.executeSql("直接写sql");
//        tEnv.sqlQuery("select * from "+table+" where id='sensor_1'").execute().print();


        // 2. 查询以注册的表
        tEnv.createTemporaryView("sensor",table); // 先注册一张表

        tEnv.sqlQuery("select * from sensor where id='sensor_1'").execute().print();

    }
}
